package com.intct.util;

import com.alibaba.fastjson.JSONObject;
import javafx.scene.control.Tab;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @author gufg
 * @since 2025-08-20 10:10
 */
public class HBaseUtil {
    /**
     * 创建HBase连接
     * @return
     * @throws IOException
     */
    public static Connection getConnection() throws IOException {
        // 创建连接
        return ConnectionFactory.createConnection();
    }


    /**
     * 创建HBase异步连接
     * @return
     * @throws Exception
     */
    public static AsyncConnection getAsyncConnection() throws Exception {
         return ConnectionFactory.createAsyncConnection().get();
    }

    /**
     * 关闭HBase异步连接
     * @param asyncConnection
     * @throws IOException
     */
    public static void closeAsycnConnection(AsyncConnection asyncConnection) throws IOException {
        if (asyncConnection != null) {
            asyncConnection.close();
        }
    }

    /**
     * 创建HBase表
     * @param connection
     * @param namesapce
     * @param tableName
     * @param familys
     * @throws IOException
     */
    public static void createTable(Connection connection, String namesapce, String tableName, String... familys) throws IOException {
        // 创建Admin实例
        Admin admin = connection.getAdmin();

        // 判断表不存在
        if (!admin.tableExists(TableName.valueOf(namesapce, tableName))) {
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(namesapce, tableName));

            for (String family : familys) {
                ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family));
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
            }
            // 创建表
            try {
                admin.createTable(tableDescriptorBuilder.build());
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        // 关闭Admin实例
        admin.close();
    }

    /**
     * 删除HBase表
     * @param connection
     * @param namesapce
     * @param tableName
     * @throws IOException
     */
    public static void dropTable(Connection connection, String namesapce, String tableName) throws IOException {
        // 创建Admin实例
        Admin admin = connection.getAdmin();

        // 判断表不存在
        if (!admin.tableExists(TableName.valueOf(namesapce, tableName))) {

            // 创建表
            try {
                // 表未禁用
                if (!admin.isTableDisabled(TableName.valueOf(namesapce, tableName))) {
                    // 禁用
                    admin.disableTable(TableName.valueOf(namesapce, tableName));
                }

                admin.deleteTable(TableName.valueOf(namesapce, tableName));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        // 关闭Admin实例
        admin.close();
    }

    /**
     * 保存数据到HBase表
     * @param connection
     * @param namesapce
     * @param tableName
     * @param put
     * @throws IOException
     */
    public static void save(Connection connection, String namesapce, String tableName, Put put) throws IOException {
        // 创建Table实例
        Table table = connection.getTable(TableName.valueOf(namesapce, tableName));

        // 保存数据
        try {
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 关闭Table实例
        table.close();
    }

    public static String getRowkey(Connection connection, String namesapce, String tableName, String rowkey) throws IOException {
        // 创建Table实例
        Table table = connection.getTable(TableName.valueOf(namesapce, tableName));

        // 声明
        JSONObject jsonObj = new JSONObject();

        // 保存数据
        try {
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = table.get(get);

            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                jsonObj.put(Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
            }

        } catch (IOException e) {
            e.printStackTrace();
        }

        // 关闭Table实例
        table.close();

        return jsonObj.toJSONString();
    }


    /**
     * 根据Rowkey异步查询表
     * @param asyncConnection
     * @param namesapce
     * @param tableName
     * @param rowkey
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static String getRowkey(AsyncConnection asyncConnection, String namesapce, String tableName, String rowkey) {
        AsyncTable<AdvancedScanResultConsumer> table = asyncConnection.getTable(TableName.valueOf(namesapce, tableName));

        // 声明
        JSONObject jsonObj = new JSONObject();
        try {
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = table.get(get).get();

            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                jsonObj.put(Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return jsonObj.toJSONString();
    }

    /**
     * 关闭HBase连接
     * @param connection
     * @throws IOException
     */
    public static void closeConnection(Connection connection) throws IOException {
        if (connection != null) {
            connection.close();
        }
    }

    public static JSONObject readDimAsync(AsyncConnection asyncConn,String namespace, String tableName, String rowKey) {
        try {
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncConn.getTable(tableNameObj);
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if (cells != null && cells.size() > 0) {
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName, columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }
}
